package com.abyss.task;


import com.abyss.bean.CleanBean;
import com.abyss.bean.StockBean;
import com.abyss.config.QuotConfig;
import com.abyss.function.KeyFunction;
import com.abyss.function.MinStockWindowFunction;
import com.abyss.inter.ProcessDataInterface;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

/**
 * @Date 2020/9/16
 * 个股分时行情
 */
public class StockMinTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        //1.分时行情时间窗口60s
        //2.获取分时成交量/金额（当前窗口的总成交金额/量 - 上一窗口的总成交金额/量-》MapState）
        //3.要写入Druid=》将数据包装成json字符串-》先手动写入Kafka ->再写入Druid

        /**
         * 开发步骤：
         * 1.定义侧边流
         * 2.数据分组
         * 3.划分时间窗口
         * 4.分时数据处理（新建分时窗口函数）
         * 5.数据分流
         * 6.数据分流转换
         * 7.分表存储(写入kafka)
         */
        //接收深市行情数据
        OutputTag<StockBean> szseOpt = new OutputTag<>("szseOpt", TypeInformation.of(StockBean.class));
        //2.数据分组
        SingleOutputStreamOperator<StockBean> processData = waterData.keyBy(new KeyFunction())
                //3.划分时间窗口
                .timeWindow(Time.minutes(1))
                //4.分时数据处理（新建分时窗口函数）
                .apply(new MinStockWindowFunction())
                //5.数据分流
                .process(new ProcessFunction<StockBean, StockBean>() {
                    @Override
                    public void processElement(StockBean value, Context ctx, Collector<StockBean> out) throws Exception {

                        if (value.getSource().equals(QuotConfig.config.getProperty("sse.topic"))) {

                            out.collect(value);
                        } else { //深市行情
                            ctx.output(szseOpt, value);
                        }
                    }
                });

        //要写入Druid=》将数据包装成json字符串-》先手动写入Kafka ->再写入Druid
        // 6.数据分流转换
        //沪市sse
        SingleOutputStreamOperator<String> sseStr = processData.map(new MapFunction<StockBean, String>() {
            @Override
            public String map(StockBean value) throws Exception {
                return JSON.toJSONString(value);
            }
        });

        //深市szse
        SingleOutputStreamOperator<String> szseStr = processData.getSideOutput(szseOpt).map(new MapFunction<StockBean, String>() {
            @Override
            public String map(StockBean value) throws Exception {
                return JSON.toJSONString(value);
            }
        });

        //7.分表存储(写入kafka)
        //新建kafka生产者对象
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers",QuotConfig.config.getProperty("bootstrap.servers"));
        FlinkKafkaProducer011<String> sseKafkaProc = new FlinkKafkaProducer011<>(QuotConfig.config.getProperty("sse.stock.topic"), new SimpleStringSchema(), properties);
        FlinkKafkaProducer011<String> szseKafkaProc = new FlinkKafkaProducer011<>(QuotConfig.config.getProperty("szse.stock.topic"), new SimpleStringSchema(), properties);

        //写入kafka
        sseStr.addSink(sseKafkaProc);
        szseStr.addSink(szseKafkaProc);
    }
}